package com.youyu.core.jms.kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;

import com.alibaba.fastjson.JSONObject;
import com.youyu.core.jms.kafka.model.KafkaBizData;
import com.youyu.core.jms.kafka.partition.DefaultPartition;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.yx.conf.AppInfo;
import org.yx.log.Log;
import scala.actors.threadpool.Arrays;

/**
 * <p>Author:  dxj</p>
 * <p>Date:2017/8/25 15:22</p>
 * <p>Description:生产者</p>
 * <p>Copyright: Copyright (c) 2017</p>
 * <p>Company: youyu</p>
 */
public class Productor<Key>{
	
	public static final String NO_RECEIPT = "0";
	public static final String LEAD_RECEIPT = "1";
	public static final String ALL_RECEIPT = "-1";
	
	private static final String KEY_SERIALIZER_CLS = "kafka.serializer.StringEncoder";

	private String topic;
	
	private List<String> brokers;
	
	private String receipt;
	
	private Class<?> partitionCls;
	
	private Producer<Key, String> producer;
	
	/**
	 * 
	 * @param topic topic名称
	 * @param receipt 发送时的等待回执参数
	 */
	public Productor(String topic, String receipt){
		this.topic = topic;
        String brokerAddrs = AppInfo.get("kafka.broker.list");
        String[] brokers = brokerAddrs.split(",");
		this.brokers = Arrays.asList(brokers);
		if(NO_RECEIPT.equals(receipt) || LEAD_RECEIPT.equals(receipt) || ALL_RECEIPT.equals(receipt)){
			this.receipt = receipt;
		}else{
			throw new IllegalArgumentException("receipt参数必须取值于本类的三个公开常量");
		}
        open();
	}
	
	/**
	 * @param topic topic名称
	 * @param receipt 发送时的等待回执参数
	 * @param partitionCls 分区类
	 */
	public Productor(String topic, String receipt, Class<?> partitionCls){
		this(topic, receipt);
		if(partitionCls.isInterface()){
			throw new IllegalArgumentException("partitionCls 必须是类而不能是接口");
		}
		if(partitionCls.getGenericSuperclass().toString().contains("com.youyu.core.jms.kafka.partition.AbstractPartition") == false){
			throw new IllegalArgumentException("partitionCls 必须是com.youyu.core.jms.kafka.partition.AbstractPartition的子类");
		}
		this.partitionCls = partitionCls;
        open();
	}
	
	/**
	 * 发送单条消息
	 * @param key
	 * @param message
	 * @return
	 */
	public void sendOneByKey(Key key, String message){
		try{
			KeyedMessage<Key, String> data = new KeyedMessage<Key, String>(this.topic, key, message);
			this.producer.send(data);
		}catch(Exception e){
            Log.get(Productor.class).error("消息发送失败,队列："+topic+"消息["+ JSONObject.toJSONString(message)+"]");
        }finally {
            this.close();
        }
	}
	
	public void open(){
		
        Properties props = new Properties();
        /**
         * 所有broker的ip:port 的字符串，用逗号分隔
         */
        props.put("metadata.broker.list",this.getBrokerAddrs());
        
        /**
         * 把发送的消息进行序列化时使用的类
         */
        props.put("serializer.class", KEY_SERIALIZER_CLS);
       
        /**
         * 发送的消息的KEY进行序列化时使用的类
         */
        props.put("key.serializer.class", KEY_SERIALIZER_CLS);
        /**
         * 标记消息发送时是否需要服务端给一个回执
         *  1 发送之后，leader给回执即可(较安全)
         *  0 发送之后不需要回执 (此时会丢失数据)
         * -1 发送之后，所有节点都给回执才可以(最安全，但耗费资源多)
         * 默认是0.
         */
        props.put("request.required.acks", this.receipt);
        
        /**
         * 用于判断一条消息应该发送到哪个分区的类
         * 这是一个可选的配置，当topic无分区时，不需要写
         * TODO 当topic存在分区时如果不写会怎么样还有待学习
         */
        if(this.partitionCls != null){
        	props.put("partitioner.class",this.partitionCls.getName());
        }
        ProducerConfig config = new ProducerConfig(props);
        this.producer = new Producer<Key, String>(config);
	}
	
	public void close(){
		if(this.producer != null){
			this.producer.close();
		}
	}

    /**
     * 获取broker列表
     * @return
     */
	private String getBrokerAddrs(){
		StringBuilder builder = new StringBuilder();
		for(int i = 0, len = this.brokers.size(); i < len; i++){
			if(i != 0){
				builder.append(",");
			}
			builder.append(this.brokers.get(i));
		}
		return builder.toString();
	}
}
